package com.liu.producer;

import com.liu.agreement.Agreement;
import com.liu.common.Constants;
import com.liu.producer.message.Message;
import com.liu.message.MessageTypeEnum;
import com.liu.producer.codec.MessageCodec;
import com.liu.producer.handler.ResponseMessageHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;


/**
 * 单例保证只有一个Client
 *
 * 注册解析器
 * MessageResolverProvider.registerResolver(resolver);
 *
 * 注册序列化器
 * SerializationProvider.registerSerialization(serialization);
 *
 * 继承SimpleChannelInboundHandler<>类添加处理器
 * @param
 * @BelongsProject: MessageQueueDEMO
 * @BelongsPackage: com.liu.producer
 * @author: L
 * @Date: 2023/10/14 16:20
 * @return
 */
public class Client {

    private String host;
    private Integer port;
    private static volatile Client instance;
    private static volatile Channel channel;

    private EventLoopGroup eventExecutors;

    public Boolean send(Message msg) {
        Agreement agreement = new Agreement(Constants.MAGIC_NUMBER,
                Constants.VERSION,
                MessageTypeEnum.PUSH_ORDINARY_REQUEST,
                msg.getSerializationType(),
                msg.getTopic(),
                msg.getMessageID(),
                msg.getBody());
        getChannel().writeAndFlush(agreement);
        DefaultPromise<Object> promise = new DefaultPromise<>(getChannel().eventLoop());
        ResponseMessageHandler.PROMISE_MAP.put(agreement.getMessageID(), promise);
        try {
            promise.sync();
            if (promise.isSuccess()) {
                // 调用正常
                System.out.println(promise);
                return true;
            } else {
                // 调用失败
                throw new RuntimeException(promise.cause());
            }

        } catch (InterruptedException e) {
            getChannel().close();
            throw new RuntimeException(e);
        } finally {
            getChannel().close();
        }
    }

    public Client(String host, Integer port) {
        this.host = host;
        this.port = port;
    }

    public static Client getClient(String host, Integer port) {
        if (instance == null) {
            synchronized (Client.class) {
                if (instance == null) {
                    instance = new Client(host, port);
                }
            }
        }
        return instance;
    }


    private Channel getChannel() {
        if (channel == null) {
            synchronized (Client.class) {
                if (channel == null) {
                    init();
                }
                return channel;
            }
        }
        return channel;
    }

    private void init() {
        LoggingHandler loggingHandler = new LoggingHandler(LogLevel.INFO);
        LengthFieldBasedFrameDecoder lengthFieldBasedFrameDecoder = new LengthFieldBasedFrameDecoder(2048, 19, 4, 0, 0);
        MessageCodec messageCodec = new MessageCodec();
        ResponseMessageHandler responseMessageHandler = new ResponseMessageHandler();
        eventExecutors = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventExecutors);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel ch) {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(lengthFieldBasedFrameDecoder);
                pipeline.addLast(loggingHandler);
                pipeline.addLast(messageCodec);
                pipeline.addLast(responseMessageHandler);
            }
        });
        try {
            ChannelFuture channelFuture = bootstrap.connect(host, port);
            channelFuture.sync();
            channel = channelFuture.channel();
            channel.closeFuture().addListener(futur -> {
                eventExecutors.shutdownGracefully();
            });
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
